package geektime.spring.data.reactive.simpler2dbcdemo;

import geektime.spring.data.reactive.simpler2dbcdemo.converter.MoneyReadCoverter;
import geektime.spring.data.reactive.simpler2dbcdemo.converter.MoneyWriteConverter;
import geektime.spring.data.reactive.simpler2dbcdemo.model.Coffee;
import io.r2dbc.h2.H2ConnectionConfiguration;
import io.r2dbc.h2.H2ConnectionFactory;
import io.r2dbc.spi.ConnectionFactory;
import java.util.Arrays;
import java.util.concurrent.CountDownLatch;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;
import org.springframework.data.convert.CustomConversions;
import org.springframework.data.convert.CustomConversions.StoreConversions;
import org.springframework.data.domain.PageRequest;
import org.springframework.data.domain.Sort;
import org.springframework.data.domain.Sort.Direction;
import org.springframework.data.r2dbc.config.AbstractR2dbcConfiguration;
import org.springframework.data.r2dbc.dialect.Dialect;
import org.springframework.data.r2dbc.function.DatabaseClient;
import org.springframework.data.r2dbc.function.convert.R2dbcCustomConversions;

@Slf4j
@SpringBootApplication
public class SimpleR2dbcDemoApplication extends AbstractR2dbcConfiguration implements
    ApplicationRunner {

  @Autowired
  private DatabaseClient databaseClient;

  public static void main(String[] args) {
    SpringApplication.run(SimpleR2dbcDemoApplication.class, args);
  }

  @Override
  public void run(ApplicationArguments args) throws Exception {
    CountDownLatch cdl = new CountDownLatch(2);

    databaseClient.execute()
        .sql("select * from t_coffee")
        .as(Coffee.class)
        .fetch()
        .first()
        .doFinally(s -> cdl.countDown())
//        .subscribeOn(Schedulers.elastic())
        .subscribe(c -> log.info("Fetch execute(){}", c));

    databaseClient.select()
        .from("t_coffee")
        .orderBy(Sort.by(Direction.DESC, "id"))
        .page(PageRequest.of(0, 3))
        .as(Coffee.class)
        .fetch()
        .all()
        .doFinally(s -> cdl.countDown())
//        .subscribeOn(Schedulers.elastic())
        .subscribe(c -> log.info("Fetch select (){}", c));

    log.info("After starting.");
    cdl.countDown();
  }

  @Bean
  public ConnectionFactory connectionFactory() {
    return new H2ConnectionFactory(
        H2ConnectionConfiguration.builder().inMemory("testdb").username("sa").build());
  }

  @Bean
  public R2dbcCustomConversions r2dbcCustomConversions() {
    Dialect dialect = getDialect(connectionFactory());
    CustomConversions.StoreConversions storeConversions = StoreConversions
        .of(dialect.getSimpleTypeHolder());
    return new R2dbcCustomConversions(storeConversions,
        Arrays.asList(new MoneyReadCoverter(), new MoneyWriteConverter()));
  }

}













